package com.example.rocketmqdemo.transaction.infr.transactionexecutor;

import com.alibaba.cola.exception.SysException;
import com.example.rocketmqdemo.transaction.domain.gateway.TransactionMessageGateway;
import com.example.rocketmqdemo.transaction.domain.transactionexecutor.LocalTransactionExecutor;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.stereotype.Component;

@Component
public class TransactionMessageGatewayImpl implements TransactionMessageGateway {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Override
    public void send(String destination, Object body, LocalTransactionExecutor<?> localTransactionExecutor) {
        Message message = new GenericMessage<>(body);
        try {
            TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(destination, message, localTransactionExecutor);
            if (!SendStatus.SEND_OK.equals(transactionSendResult.getSendStatus())) {
                String msg = String.format("事务消息发送失败. destination: %s, body: %s, result: %s", destination, body, transactionSendResult);
                throw new SysException(msg);
            }
        } catch (MessagingException e) {
            String msg = String.format("事务消息发送失败. destination: %s, body: %s", destination, body);
            throw new SysException(msg, e);
        }
    }
}
